package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String, String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public AppLogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                      OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;

    }

    @Override
    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
        //第1步、解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);

        //第2步、错误日志
        if(null != jsonObject.get("err")){
            //侧边流输出
            ctx.output(errorTag, value);
            //错误数据已经单独保存，无需在保存，直接删除
            jsonObject.remove("err");
        }

        //第3步、启动日志
        if(null != jsonObject.get("start")){
            //侧边流输出
            ctx.output(startTag, jsonObject.toJSONString());
        }
        //第4步、页面日志
        else {
            //获取曝光和动作数据
            JSONArray displaysJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");

            //删除曝光和动作
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            //输出页面日志
            out.collect(jsonObject.toJSONString());

            //第5步、曝光日志
            if(null != displaysJsonArray && !displaysJsonArray.isEmpty()){
                for (int i=0;i<displaysJsonArray.size();i++){
                    JSONObject displayJsonObject = displaysJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("display", displayJsonObject);
                    //输出
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }
            //将上述添加曝光数据字段值删除
            jsonObject.remove("display");

            //第6步、动作日志
            if(null != actionsJsonArray && !actionsJsonArray.isEmpty()){
                for (int i=0;i<actionsJsonArray.size();i++){
                    JSONObject actionJsonObject = actionsJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("action", actionJsonObject);
                    //输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }
}
